package com.atguigu1.core.persist

import java.util.Date

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @description: checkpoint需要落盘,需要指定检查点保存路径
 *               检查点路径保存的文件数据当作业执行完毕后,不会被删除
 *               一般保存都是在分布式存储系统中hdfs
 * @time: 2021-03-12 11:45
 * @author: baojinlong
 **/
object Spark02CheckPoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("persistDemo")
    // 设置rdd分区数字
    val sparkContext = new SparkContext(conf)
    sparkContext.setCheckpointDir("cpxx")
    val rdd: RDD[String] = sparkContext.makeRDD(Seq("hello scala", "hello spark", "hello windy"))
    val flatRdd: RDD[String] = rdd.flatMap(_.split(" "))
    val mapRdd: RDD[(String, Int)] = flatRdd.map({
      word =>
        print("word=" + word)
        (word, 1)
    })
    // 保存到内存中,cache默认持久化的操作,只能将数据保存到内存中,如果想要保存到磁盘文件中需要更改缓存策略,持久化操作必须在行动算子完成后才触发
    mapRdd.checkpoint
    val reduceRdd: RDD[(String, Int)] = mapRdd.reduceByKey(_ + _)
    reduceRdd.collect.foreach(println)
    println("============================")

    println("----------------------")
    val groupRdd: RDD[(String, Iterable[Int])] = mapRdd.groupByKey()
    groupRdd.collect.foreach(println)
    sparkContext.stop()
  }

  def showLog(d: Date, log: String): Unit = {
    println("时间：" + d, ",日志信息：" + log)
  }

}
